package cn.dmrliu.edu.realtime.app.dwd.db;

import cn.dmrliu.edu.realtime.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradePaymentSuc {
    public static void main(String[] args) {
        // TODO 1.基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // TODO 2.检查点的设置

        // TODO 3.从kafka读取 dwd_trade_order_detail 数据，创建动态表
        tableEnv.executeSql("CREATE TABLE dwd_trade_order_detail (\n" +
                "  id string,\n" +
                "  order_id string,\n" +
                "  user_id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  final_amount string,\n" +
                "  province_id string,\n" +
                "  date_id string,\n" +
                "  ts string,\n" +
                "  row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))),\n" +
                "  watermark for row_time as row_time" +
                ")" + KafkaUtil.getKafkaDDL("edu_dwd_trade_order_detail", "edu_dwd_trade_payment_suc_group_1"));

//        tableEnv.executeSql("select * from dwd_trade_order_detail").print();

        // TODO 4.筛选支付数据
        tableEnv.executeSql("create table edu_db(" +
                "  `database` String,\n" +
                "  `table` String,\n" +
                "  `type` String,\n" +
                "  `data` map<String, String>,\n" +
                "  `old` map<String, String>,\n" +
                "  `proc_time` as PROCTIME(),\n" +
                "  `ts` string,\n" +
                "  row_time as TO_TIMESTAMP(FROM_UNIXTIME(cast(ts as bigint))),\n" +
                "  watermark for row_time as row_time" +
                ")" + KafkaUtil.getKafkaDDL("edu_dwd_trade_payment_suc_group_2"));
        Table paymentSuc = tableEnv.sqlQuery("select \n" +
                " `data`['id'] id,\n" +
                " `data`['order_id'] order_id,\n" +
                " `data`['payment_type'] payment_type,\n" +
                " `data`['callback_time'] callback_time,\n" +
                " row_time,\n" +
                " ts,\n" +
                " `proc_time`\n" +
                "from edu_db\n" +
                "where `table` = 'payment_info'\n");
        tableEnv.createTemporaryView("payment_suc", paymentSuc);

//        tableEnv.executeSql("select * from payment_suc").print();

        // TODO 5.join
        Table joinTable = tableEnv.sqlQuery("select\n" +
                "  od.id order_detail_id,\n" +
                "  od.order_id,\n" +
                "  od.user_id,\n" +
                "  od.course_id,\n" +
                "  od.course_name,\n" +
                "  od.province_id,\n" +
                "  ps.payment_type,\n" +
                "  ps.callback_time,\n" +
                "  od.origin_amount,\n" +
                "  od.coupon_reduce,\n" +
                "  od.final_amount payment_amount,\n" +
                "  ps.ts\n" +
                "from dwd_trade_order_detail od , payment_suc ps\n" +
                "where od.order_id = ps.order_id\n" +
                "and od.row_time >= ps.row_time - INTERVAL '15' MINUTE \n" +
                "and od.row_time <= ps.row_time + INTERVAL '5' SECOND");
        tableEnv.createTemporaryView("join_table", joinTable);

//        tableEnv.executeSql("select * from join_table").print();
        // TODO 疑问：一直不打印结果，添加了web依赖后才显示结果


        // TODO 6.写入kafka
        tableEnv.executeSql("create table dwd_trade_payment_suc(\n" +
                "  order_detail_id string,\n" +
                "  order_id string,\n" +
                "  user_id string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  province_id string,\n" +
                "  payment_type string,\n" +
                "  callback_time string,\n" +
                "  origin_amount string,\n" +
                "  coupon_reduce string,\n" +
                "  payment_amount string,\n" +
                "  ts string,\n" +
                "  PRIMARY KEY (order_detail_id) NOT ENFORCED\n" +
                ")"
                + KafkaUtil.getUpsertKafkaDDL("edu_dwd_trade_payment_sec"));
        tableEnv.executeSql("insert into dwd_trade_payment_suc select * from join_table");
    }
}
